package com.tm.test;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @author Rocky
 * @create 2022-10-11 18:48
 * 顺序消息分两种：全局顺序和局部顺序
 * 全局顺序：
 * 保证某个topic下的所有消息都要保证消费顺序和产生顺序相同。
 * 局部顺序：
 * 只需要保证某一组消息被取出时是顺序取出的即可。
 * 要实现全局顺序，需要将生产者和消费者都设置为单线程，这样可以绝对保证数据的顺序，
 * 但是牺牲了高并发和高吞吐量。所以一般来说需要根据业务来实现某个模块或某个功能的局部顺序消息即可。
 */
public class RocketMQTest4 {
    //顺序消费消息
    public static void main(String[] args) throws Exception {
        //创建消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("zkj66");
        //设置nameServer
        consumer.setNamesrvAddr("192.168.0.3:9876");
        //设置消费消息的顺序
        //有现成的枚举,LAST就是先进先出
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //设置要读取的消息的topic和tag
        consumer.subscribe("order","add");
        //顺序读取消息
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage (List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                //设置消息读取后自动删除
                consumeOrderlyContext.setAutoCommit(true);
                //list中就是读取到的所有消息
                for (MessageExt messageExt : list) {
                    String msgId = messageExt.getMsgId();
                    String string = new String(messageExt.getBody());
                    JSONObject jsonObject = JSONObject.parseObject(string);
                    System.out.println(msgId);
                    System.out.println(string);
                    System.out.println(jsonObject);
                }
                //返回成功标识
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        //启动消费者
        consumer.start();
        System.out.println("Consumer启动成功！");
        //    //消费者需要关闭吗？ 不用
        //    //1、我们在项目中使用消息消费者，是需要持续不断的读取MQ中的消息，
        //    //所以不用关闭
        //    //2、在读取消息时，是自动开启另一线程，和当前代码不是同时执行
        //    //可能造成，上面的consumeMessage方法还没走完，却已经consumer.shutdown了
        //    //consumer.shutdown();
    }


}
